package com.ruoyi.web.core.config.mqtt.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.ruoyi.common.constant.RedisConstans;
import com.ruoyi.common.core.domain.HatDeviceLog;
import com.ruoyi.common.utils.spring.SpringUtils;
import lombok.SneakyThrows;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @Author wy
 * @Description 设备端发布数据回调
 * @Date 2023/11/22 16:08
 * @Version 1.0
 */
@Log4j2
public class MqttConsumerDeviceCallBack implements MqttCallback {
    /**
     * 不能使用autowird自动装配
     */
    private RedisTemplate  redisTemplate;

    private MqttConsumerConfig mqttConsumerConfig;


    /**
     * 存放的数据队列
     */
    private static ConcurrentLinkedQueue<HatDeviceLog>   thisLinkedQueue=new ConcurrentLinkedQueue<>();

    /**
     * 队列的值
     * @return
     */
    public static ConcurrentLinkedQueue<HatDeviceLog> getThisLinkedQueue() {
        return thisLinkedQueue;
    }

    public MqttConsumerDeviceCallBack(MqttConsumerConfig mqttConsumerConfig) {
        this.mqttConsumerConfig = mqttConsumerConfig;
        this.redisTemplate= SpringUtils.getBean("redisTemplate");
        System.out.println(this.redisTemplate);
    }

    /**
     *  与服务器断开连接，可重连
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("---MqttConsumerDeviceCallBack与服务器断开连接----");
        log.error(throwable);
        this.mqttConsumerConfig.connect();
    }

    /**
     * 消息到达的回调（区域设备的数据，现在的位置）
     * @param topic
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
//        System.out.println(String.format("接收消息主题 : %s",topic));
//        System.out.println(String.format("接收消息内容 : %s",new String(mqttMessage.getPayload())));
//        System.out.println(System.currentTimeMillis());
        //{"deviceId":104,"deviceName":"重庆安全帽CQ1004","deviceCode":"D202311200004","lon":106.497298,"lat":29.547298,"sysOrgCode":"A01A06A01"}

        String  info=new String(mqttMessage.getPayload());
        HatDeviceLog hatDeviceLog = JSONUtil.toBean(info, HatDeviceLog.class, true);
        hatDeviceLog.setCreateTime(new Date());

        //把数据先更新到redis的hash中
        this.redisTemplate.opsForHash().put(RedisConstans.HAT_DEVICE_HASH_KEY,hatDeviceLog.getDeviceCode(),hatDeviceLog);

        //然后再把数据，按设备编号进行按年月分类存储
        String ym=LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMM"));
        //hat_device_log:设备编号:年月
        String redisKey= StrUtil.format("{}{}:{}",RedisConstans.HAT_DEVICE_LOG,hatDeviceLog.getDeviceCode(),ym);
        this.redisTemplate.opsForList().leftPush(redisKey,hatDeviceLog);

        thisLinkedQueue.add(hatDeviceLog);
    }


    /**
     * 消息发布成功的回调
     * @param iMqttDeliveryToken
     */
    @SneakyThrows
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info(iMqttDeliveryToken.getMessage());
    }



}
